package com.wujian.love.redis.stream;

import com.wujian.love.contants.ClassContants;
import com.wujian.love.redis.RedisService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import redis.clients.jedis.StreamEntry;
import redis.clients.jedis.StreamEntryID;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @ClassName: MessageStreamConsumerService
 * @Description: 消息的消费者
 * @Author: wuj
 * @Date: 2021-02-03 11:00
 **/
@Service
public class MessageStreamConsumerService implements MessageListener {


    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedisService redisService;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        doBusiness(message);
    }


    /**
     * 打印 message body 内容
     *
     * @param message
     */
    public void doBusiness(Message message) {
        RedisSerializer<?> serializer = redisTemplate.getValueSerializer();
        // message.getBody()是Redis的值，需要用redis的valueSerializer反序列化
        String topic = redisTemplate.getStringSerializer().deserialize(message.getChannel()).toString();
        String key = serializer.deserialize(message.getBody()).toString();
        System.out.println("topic:" + topic + "   key:" + key);
        //null 则为 > 重头读起，也可以为$接受新消息，还可以是上一次未读完的消息id
        Map<String, StreamEntryID> map = new HashMap<>();
        map.put(key, null);
        Map.Entry e = null;
        for(Map.Entry c:map.entrySet()){
            e=c;
        }
        //noAck为false的话需要手动ack，true则自动ack. commsumer新建的方式为xreadgroup。
        List<Map.Entry<String, List<StreamEntry>>> list = redisService.xreadGroup(ClassContants.REDIS_STREAM_GROUP_NAME, ClassContants.REDIS_STREAM_CONSUMER_NAME, 10, 1000, false, e);
        if (list != null && list.size() > 0) {
            for (Map.Entry<String, List<StreamEntry>> stringListEntry : list) {
                List<StreamEntry> value = stringListEntry.getValue();
                String streamKey = stringListEntry.getKey();
                System.out.println("stream value:" + value);
                System.out.println("stream streamKey:" + streamKey);
                for (StreamEntry streamEntry : value) {
                    System.out.println("streamEntry id:" + streamEntry.getID());
                    System.out.println("streamEntry fields:" + streamEntry.getFields());
                    System.out.println("消费消息");
                    Long xack = redisService.xack(key, streamEntry);
                    System.out.println("xack:"+xack);
                }

            }
        } else {
            System.out.println("没有数据");
        }



        //System.out.println("消息消费成功");


    }


}
